package org.qpcr.community.device.service;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.qpcr.community.device.config.MessageQueueConfig;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 响应式消息队列服务
 * 提供响应式编程风格的消息发送和接收功能
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class ReactiveMessageQueueService {

    private final StringRedisTemplate stringRedisTemplate;
    private final MessageQueueConfig config;
    private final AtomicLong messageCounter = new AtomicLong(0);


    /**
     * 响应式发送消息到指定队列
     *
     * @param queueName 队列名称
     * @param message 消息内容
     * @return 发送结果的Mono
     */
    public Mono<Long> sendMessage(String queueName, String message) {
        return Mono.fromCallable(() -> {
                log.debug("准备发送消息到队列: {}", queueName);
                return message;
            })
            .flatMap(msg -> {
                long messageId = messageCounter.incrementAndGet();
                log.info("发送消息 [{}] 到队列: {}, 内容长度: {}", messageId, queueName, msg.length());

                return Mono.fromCallable(() ->
                    stringRedisTemplate.opsForList().rightPush(queueName, msg)
                )
                .subscribeOn(Schedulers.boundedElastic())
                .map(result -> {
                    log.info("消息 [{}] 已成功发送到队列: {}, 结果: {}", messageId, queueName, result);
                    return result;
                })
                .onErrorResume(error -> {
                    log.error("消息 [{}] 发送到队列 {} 失败: {}", messageId, queueName, error.getMessage(), error);
                    return Mono.error(new RuntimeException(
                        String.format("发送消息到队列 %s 失败: %s", queueName, error.getMessage()),
                        error
                    ));
                });
            })
            .doOnSuccess(result ->
                log.debug("消息发送成功，队列: {}, 结果: {}", queueName, result)
            )
            .doOnError(error ->
                log.error("消息发送失败，队列: {}, 错误: {}", queueName, error.getMessage(), error)
            );
    }

    /**
     * 响应式发送消息到指定队列，带重试机制
     *
     * @param queueName 队列名称
     * @param message 消息内容
     * @return 发送结果的Mono（使用默认重试配置）
     */
    public Mono<Long> sendMessageWithRetry(String queueName, String message) {
        return sendMessageWithRetry(queueName, message, config.getDefaultRetryCount(), config.getDefaultRetryDelay());
    }

    /**
     * 响应式发送消息到指定队列，带重试机制
     *
     * @param queueName 队列名称
     * @param message 消息内容
     * @param maxRetries 最大重试次数
     * @param retryDelay 重试延迟
     * @return 发送结果的Mono
     */
    public Mono<Long> sendMessageWithRetry(String queueName, String message, int maxRetries, Duration retryDelay) {
        return sendMessage(queueName, message)
            .retryWhen(reactor.util.retry.Retry.backoff(maxRetries, retryDelay)
                .filter(error -> error instanceof RuntimeException)
                .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> {
                    log.error("消息发送重试 {} 次后仍然失败，队列: {}", maxRetries, queueName);
                    return new RuntimeException("消息发送重试失败", retrySignal.failure());
                }));
    }

    /**
     * 响应式批量发送消息到指定队列
     *
     * @param queueName 队列名称
     * @param messages 消息列表
     * @return 发送结果的Flux
     */
    public Flux<Long> sendBatchMessages(String queueName, Iterable<String> messages) {
        return Flux.fromIterable(messages)
            .flatMap(message -> sendMessage(queueName, message), config.getBatchConcurrency())
            .doOnComplete(() ->
                log.info("批量消息发送完成，队列: {}", queueName)
            )
            .doOnError(error ->
                log.error("批量消息发送失败，队列: {}, 错误: {}", queueName, error.getMessage(), error)
            );
    }

    /**
     * 响应式从队列接收消息（使用默认超时时间）
     *
     * @param queueName 队列名称
     * @return 消息内容的Mono
     */
    public Mono<String> receiveMessage(String queueName) {
        return receiveMessage(queueName, config.getDefaultMessageTimeout());
    }

    /**
     * 响应式从队列接收消息
     *
     * @param queueName 队列名称
     * @param timeout 超时时间
     * @return 消息内容的Mono
     */
    public Mono<String> receiveMessage(String queueName, Duration timeout) {
        return Mono.fromCallable(() ->
            stringRedisTemplate.opsForList().leftPop(queueName, timeout)
        )
        .subscribeOn(Schedulers.boundedElastic())
        .filter(message -> message != null)
        .doOnNext(message ->
            log.debug("从队列 {} 接收到消息，长度: {}", queueName, message.length())
        )
        .doOnError(error ->
            log.error("从队列 {} 接收消息失败: {}", queueName, error.getMessage(), error)
        );
    }

    /**
     * 响应式监听队列消息（使用默认轮询间隔）
     *
     * @param queueName 队列名称
     * @return 消息内容的Flux
     */
    public Flux<String> listenQueue(String queueName) {
        return listenQueue(queueName, config.getDefaultPollInterval());
    }

    /**
     * 响应式监听队列消息
     *
     * @param queueName 队列名称
     * @param pollInterval 轮询间隔
     * @return 消息内容的Flux
     */
    public Flux<String> listenQueue(String queueName, Duration pollInterval) {
        return Flux.interval(pollInterval)
            .onBackpressureBuffer(config.getBackpressureBufferSize()) // 使用配置的背压缓冲区大小
            .flatMap(tick -> receiveMessage(queueName, Duration.ofMillis(100)), 1)
            .doOnNext(message ->
                log.debug("监听到队列 {} 的消息: {}", queueName, message)
            )
            .doOnError(error ->
                log.error("监听队列 {} 时发生错误: {}", queueName, error.getMessage(), error)
            )
            .onBackpressureDrop(dropped ->
                log.warn("由于背压，丢弃轮询tick: {}", dropped)
            );
    }

    /**
     * 响应式监听队列消息（改进版本，使用背压感知的轮询）
     *
     * @param queueName 队列名称
     * @param pollInterval 轮询间隔
     * @return 消息内容的Flux
     */
    public Flux<String> listenQueueWithBackpressureControl(String queueName, Duration pollInterval) {
        return Flux.defer(() -> {
                log.info("开始监听队列: {}, 轮询间隔: {}, 背压缓冲区大小: {}",
                    queueName, pollInterval, config.getBackpressureBufferSize());
                return Flux.interval(pollInterval)
                    .onBackpressureBuffer(config.getBackpressureBufferSize(), dropped ->
                        log.warn("背压缓冲区已满，丢弃tick: {}", dropped)
                    )
                    .flatMap(tick -> {
                        log.debug("轮询队列: {}, tick: {}", queueName, tick);
                        return receiveMessage(queueName, Duration.ofMillis(100))
                            .onErrorResume(error -> {
                                log.warn("轮询队列 {} 时发生错误，继续下一个tick: {}", queueName, error.getMessage());
                                return Mono.empty();
                            });
                    }, 1) // 限制并发度为1，避免过度并发
                    .doOnNext(message ->
                        log.debug("监听到队列 {} 的消息，长度: {}", queueName, message.length())
                    )
                    .doOnError(error ->
                        log.error("监听队列 {} 时发生错误: {}", queueName, error.getMessage(), error)
                    )
                    .doOnComplete(() ->
                        log.info("队列 {} 监听完成", queueName)
                    );
            })
            .retryWhen(reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(1))
                .filter(error -> error instanceof RuntimeException)
                .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> {
                    log.error("队列 {} 监听重试3次后仍然失败", queueName);
                    return new RuntimeException("队列监听失败", retrySignal.failure());
                }));
    }

    /**
     * 获取队列长度
     *
     * @param queueName 队列名称
     * @return 队列长度的Mono
     */
    public Mono<Long> getQueueSize(String queueName) {
        return Mono.fromCallable(() ->
            stringRedisTemplate.opsForList().size(queueName)
        )
        .subscribeOn(Schedulers.boundedElastic())
        .onErrorReturn(0L);
    }

    /**
     * 清空指定队列
     *
     * @param queueName 队列名称
     * @return 操作结果的Mono
     */
    public Mono<Boolean> clearQueue(String queueName) {
        return Mono.fromCallable(() -> {
                stringRedisTemplate.delete(queueName);
                return true;
            })
            .subscribeOn(Schedulers.boundedElastic())
            .doOnSuccess(result ->
                log.info("队列 {} 已清空", queueName)
            )
            .doOnError(error ->
                log.error("清空队列 {} 失败: {}", queueName, error.getMessage(), error)
            );
    }
}
